package com.mr.controller;

// 添加缺失的导入
import com.mr.job.Job;
import com.mr.job.JobTracker;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import com.mr.example.WordCountMapper;
import com.mr.example.WordCountReducer;

import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.server.ResponseStatusException;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CompletableFuture;
import java.nio.file.*;
import java.io.IOException;
import java.util.stream.*;

@RestController
public class JobController {
    // 添加时间记录字段
    private final Map<String, Long> jobStartTimes = new ConcurrentHashMap<>();
    private final Map<String, Long> jobEndTimes = new ConcurrentHashMap<>();
    
    // 首页映射为状态接口
    @GetMapping("/status")
    public String status() {
        return "MR Framework is running! Use /submit to submit jobs";
    }
    
    private final JobTracker tracker = new JobTracker();
    private final Map<String, JobStatus> jobs = new ConcurrentHashMap<>();

     // 提交MapReduce作业
    @PostMapping("/submit")
    public String submitJob(@RequestBody JobRequest request) {
        String jobId = UUID.randomUUID().toString();
        jobs.put(jobId, JobStatus.PROCESSING);
        jobStartTimes.put(jobId, System.currentTimeMillis()); // 记录开始时间
        
        CompletableFuture.runAsync(() -> {
            try {
                Job<String, String, String, Integer, String, Integer> job = new Job<>();
                job.setInputPath(request.getInputPath());
                job.setMapThreads(request.getMapThreads());
                job.setReduceThreads(request.getReduceThreads()); 
                job.setMapperClass(WordCountMapper.class); // 设置Mapper
                job.setReducerClass(WordCountReducer.class); // 设置Reducer
                tracker.run(job, jobId); // 提交给JobTracker执行
                jobs.put(jobId, JobStatus.COMPLETED);
                jobEndTimes.put(jobId, System.currentTimeMillis()); // 记录结束时间
            } catch (Exception e) {
                // 错误信息记录
                jobs.put(jobId, JobStatus.FAILED);
                jobEndTimes.put(jobId, System.currentTimeMillis()); // 失败时也记录结束时间
                System.err.println("Job failed: " + e.getMessage());
                e.printStackTrace();
            }
        });
        return jobId;
    }

    // 状态接口
    @GetMapping("/status/{jobId}")
    public String getJobStatus(@PathVariable String jobId) {
        return jobs.getOrDefault(jobId, JobStatus.NOT_FOUND).name();
    }

    // 耗时获取接口
    @GetMapping("/time/{jobId}")
    public long getProcessingTime(@PathVariable String jobId) {
        Long start = jobStartTimes.get(jobId);
        Long end = jobEndTimes.getOrDefault(jobId, System.currentTimeMillis());
        return start == null ? 0 : (end - start);
    }

    // 内部状态枚举
    enum JobStatus { PROCESSING, COMPLETED, FAILED, NOT_FOUND }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    static class WordCount {
        private String word;
        private int count;
    }

    // 获取作业结果
    @GetMapping("/result/{jobId}")
    public List<WordCount> getResult(@PathVariable String jobId) throws IOException {
        if (jobs.get(jobId) != JobStatus.COMPLETED) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "任务未完成");
        }
        
        Path outputPath = Paths.get(tracker.getJobOutputDir(jobId));
        List<WordCount> results = Files.walk(outputPath) //指定的输出目录路径中读取文件
                .filter(Files::isRegularFile)
                .flatMap(path -> { // 解析结果文件
                    try {
                        return Files.lines(path);
                    } catch (IOException e) {
                        System.err.println("读取结果文件失败: " + path);
                        return Stream.empty();
                    }
                })
                .map(line -> {
                    String[] arr = line.split("\t");
                    if (arr.length != 2) {
                        System.err.println("无效结果格式: " + line);
                        return null;
                    }
                    try {
                        String countStr = arr[1].replaceAll("[\\[\\]]", "");
                        return new WordCount(arr[0], Integer.parseInt(countStr));
                    } catch (Exception e) {
                        System.err.println("数字解析失败: " + line);
                        return null;
                    }
                })
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
                
        if (results.isEmpty()) {
            throw new ResponseStatusException(HttpStatus.NOT_FOUND, "未找到有效结果");
        }
        return results;
    }

    @GetMapping("/export/{jobId}")
    public ResponseEntity<String> exportToCSV(@PathVariable String jobId) throws IOException {
        if (jobs.get(jobId) != JobStatus.COMPLETED) {
            return ResponseEntity.status(HttpStatus.BAD_REQUEST).body("Job not completed");
        }
        
        // 使用异步处理避免阻塞
        CompletableFuture<String> csvContent = CompletableFuture.supplyAsync(() -> {
            try {
                List<WordCount> results = getResult(jobId);
                return generateCSV(results);
            } catch (Exception e) {
                throw new RuntimeException("CSV生成失败: " + e.getMessage());
            }
        });
        
        // 设置响应头
        return ResponseEntity.ok()
            .header("Content-Type", "text/csv; charset=UTF-8")
            .header("Content-Disposition", "attachment; filename=\"results_" + System.currentTimeMillis() + ".csv\"")
            .body(csvContent.join());
    }

    private String generateCSV(List<WordCount> results) {
        StringBuilder sb = new StringBuilder("\uFEFF");
        sb.append("单词,出现次数\n");
        results.stream()
            .sorted((a, b) -> b.getCount() - a.getCount())
            .forEach(wc -> {
                String escapedWord = wc.getWord().contains(",") ? 
                    "\"" + wc.getWord().replace("\"", "\"\"") + "\"" : wc.getWord();
                sb.append(escapedWord)
                  .append(",")
                  .append(wc.getCount())
                  .append("\n");
            });
        return sb.toString();
    }

}